package com.atguigu.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.EduConfig;
import com.atguigu.utils.DruidDSUtil;
import com.atguigu.utils.JedisUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Map;
import java.util.Set;

/**
 * ClassName: DimSinkFunction
 * Package: com.atguigu.func
 * Description:
 *
 * @Author Lovxy
 * @Create 2023/5/15 14:14
 * @Version 1.0
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private static DruidDataSource druidDataSource;
    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource = DruidDSUtil.getDruidDataSource();
    }

    //value:{"database":"gmall-flink","table":"comment_info","type":"insert","ts":1669162958,"xid":1111,"xoffset":13941,
    // "data":{"id":1595211185799847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,"spu_id":10,"order_id":987,"appraise":"1204","comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38","operate_time":null},
    // "sink_table":"dim_comment_info"}
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        String sinkTable = value.getString("sink_table");
        JSONObject data = value.getJSONObject("data");
        //如果是更新操作，则先写入Redis
        String type = value.getString("type");
        if ("update".equals(value.getString("type"))){
            Jedis jedis = JedisUtil.getJedis();
            String redisKey="DIM:"+sinkTable.toUpperCase()+":"+data.getString("id");
            JSONObject jsonObject = new JSONObject();

            Set<Map.Entry<String, Object>> entries = data.entrySet();
            for (Map.Entry<String, Object> entry : entries) {
                Object nextValue = entry.getValue();
                if (nextValue !=null){
                    jsonObject.put(entry.getKey().toUpperCase(), nextValue.toString());
                }
                else  jsonObject.put(entry.getKey().toUpperCase(), nextValue);

            }
            jedis.setex(redisKey,24*3600,jsonObject.toJSONString());
            jedis.close();
        }


        //获取连接
        DruidPooledConnection connection = druidDataSource.getConnection();

        //获取sql语句upsert into db.tn(id,name,sex) values('1001','zs','male')
        String upsertSql = getSql(sinkTable, data);
        System.out.println("upsertSql>>>>" + upsertSql);
        //编译sql
        PreparedStatement preparedStatement = connection.prepareStatement(upsertSql);

        //执行sql
        preparedStatement.execute();

        //提交链接
        connection.commit();

        //关闭资源
        preparedStatement.close();

        //归还链接
        connection.close();
    }

    private String getSql(String sinkTable, JSONObject data) {
        //分别取出列名和值
        Set<String> columns = data.keySet();
        Collection<Object> values = data.values();

        StringBuilder stringBuilder = new StringBuilder("upsert into " + EduConfig.HBASE_SCHEMA + "." + sinkTable + "("
                + StringUtils.join(columns, ",") + ") values('" + StringUtils.join(values, "','") + "')");

        return stringBuilder.toString();
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}